package com.huaxin.dss2.data.dao;


import com.huaxin.dss2.data.bean.DataValue;
import com.huaxin.dss2.data.dao.base.BaseDataDao;
import com.huaxin.dss2.data.exception.jdbc.DataSourceNotFoundException;
import com.huaxin.dss2.message.exception.SaveRealDataException;
import com.huaxin.dss2.util.Constant;
import com.huaxin.dss2.util.ResourcesPropertiesUtil;
import java.io.IOException;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.directory.api.util.Strings;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

/**
 * @author : luqing
 * @date : 2018/6/27 14:54
 */
@Component("hbaseDataDao")
@Slf4j
public class HBaseDataDao extends BaseDataDao {



    private static Connection connection = null;
    private static Configuration configuration = null;

    private static String tableName;
    private static String columnName;

    @Override
    public void init(){
        log.info("初始化HBase数据源");
        String clientPort = ResourcesPropertiesUtil.getString("hbase.clientPort");
        tableName = ResourcesPropertiesUtil.getString("hbase.tableName");
        columnName = ResourcesPropertiesUtil.getString("hbase.columnFamilyName");
        String parent = ResourcesPropertiesUtil.getString("znode.parent");
        String quorum = ResourcesPropertiesUtil.getString("zookeeper.quorum");

        if(configuration==null) {
            configuration = HBaseConfiguration.create();
        }
        configuration.set("zookeeper.znode.parent", parent);
        configuration.set("hbase.zookeeper.quorum", quorum);
        configuration.set("hbase.master", "10.0.0.23:2181");
        configuration.set("hbase.zookeeper.property.clientPort", clientPort);
        if(connection==null) {
            try {
                connection = ConnectionFactory.createConnection(configuration);
            } catch (IOException e) {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (IOException e1) {
                        log.error("zookeeper连接关闭异常",e);
                    }
                }
                log.error("zookeeper连接创建异常",e);
            }
        }
    }

    public HBaseDataDao() {
    }

    @Override
    public void save(String stationCode,List<DataValue> saveList, Date dataDate)
        throws DataSourceNotFoundException, SaveRealDataException {
        final String dataDateStr  = DateFormatUtils.format(dataDate,"yyyy-MM-dd HH:mm:ss");

        HTable table=null;
        try {
            // 存入hbase
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            final String daytime = dataDateStr.substring(0,11)+"00:00:00";
            Long daytimestamp =  sdf.parse(daytime).getTime();

            final long timstamp = dataDate.getTime();
            String qualifer = String.format("%06d",(timstamp-daytimestamp)/1000L);
            List<Put> putList = new ArrayList<>();
            table=(HTable)connection.getTable(TableName.valueOf(tableName));

            for(DataValue dataValue:saveList){
                String [] fullDataCode= dataValue.getDataCode().split("_");

                String dataCode = fullDataCode[2];

                if ("jiajing".equals(Constant.getProperties().getCompanyName())) {
                    //嘉靖处理iswarn key 的方式
                    if (!Strings.equals(dataCode, "iswarn")) {
                        dataCode = dataCode + "0000";
                    }
                }

                String rowkey = dataCode +String.format("%04d",Integer.parseInt(fullDataCode[1]))+fullDataCode[0]+String.valueOf(daytimestamp/1000L);

                Put put = new Put(Bytes.toBytes(rowkey));
                put.addColumn(Bytes.toBytes(columnName),Bytes.toBytes(qualifer),Bytes.toBytes(String.valueOf(dataValue.getDataValue())));
                putList.add(put);
            }

            table.put(putList);

        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new SaveRealDataException(e);
        }finally {
            if(table!=null){
                try {
                    table.close();
                }catch (Exception e){
                    log.error("hbase 表关闭失败",e);
                }
            }
        }

    }
}
